package job;

import java.util.Date;
import java.util.Properties;

import kafka.ConsumerFactory;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import play.jobs.Every;
import play.jobs.Job;
import utils.DateUtils;
import utils.PropertyUtil;
import utils.TopicUtil;
import utils.SystemParam.Topic;
import bp.NetSaleBP;

/**
 * 
 * <b>类描述：退票消费者</b><br/>
 * <b>类名称：</b>ReturnTicketConsumer<br/>
 * <b>创建人：</b>刘汝鹏</a><br/>
 * <b>关键修改：</b><br/>
 * <b>修改时间：</b><br/>
 * <b>修改人：</b><br/>
 * 
 */
@SuppressWarnings("rawtypes")
@Every("3s")
public class ReturnUnknownConsumer extends Job {
	private final ConsumerConnector consumer;
	private final String topic;

	public ReturnUnknownConsumer() {
		System.out.println("before create consumer");
		consumer = ConsumerFactory.createConsumerConnector();// 创建到zookeeper的连接
		System.out.println("after create consumer");
		this.topic = TopicUtil.getTopic("RefundUnknown");
	}


	@Override
	public void doJob() throws Exception {
		KafkaStream<String, String> kafkaStream = ConsumerFactory.createKafkaStream(topic, consumer);// 获取消息的流
		ConsumerIterator<String, String> it = kafkaStream.iterator();// 遍历消息，此函数会一直阻塞
		MessageAndMetadata<String, String> data = null;
		while (it.hasNext()) {
			data = it.next();
			// 此处做对消息的处理
			boolean flag = new NetSaleBP().returnunknown(data.message());
			if (flag == false) {
				ErrorMessageConsumer.setOnline("REUK");
				ErrorMessageConsumer.send(TopicUtil.getTopic("ErrorConsumer"), "REUK#0#" + DateUtils.formatDatetime(new Date()) + "#" + data.message());
			}
			consumer.commitOffsets();// 提交此消费者的偏移量
			// it.allDone();//调用此函数后在调用
			// consumer.shutdown线程会退出。（如若不调用此函数调用consumer.shutdown线程不会立即退出也许是处理完缓存的消息后才会退出，需要测试）
		}

		super.doJob();
	}

}
